package com.example.admin.myapplication;

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.net.Inet4Address;
import java.util.concurrent.Callable;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;

public class MainActivity extends AppCompatActivity {

    public static final String  TAGS = "RxJava";
    private Integer i = 10;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
         setContentView(R.layout.activity_main);

//          testVoid();
//          testJust();
//          tstConsumer();
//            testEmpty();
            testDefer();

    }

    /**
     *   需求场景：
     *           定时操作： 在经过x秒之后,执行Y
     *            周期性操作：
     *
     */
    private void testDefer() {  //作用 直到有（Observer）订阅时,才动态的创建（Observable）& 发送事件
            //每次订阅后，都会得到一个刚创建的最新的Observable ，以确保Observable对象里的数据都是最新的
            Observable<Integer>  observable =  Observable.defer(new Callable<ObservableSource<? extends Integer>>() {
                @Override
                public ObservableSource<? extends Integer> call() throws Exception {
                    return  Observable.just(i);
                }
            });
            i = 20;
            observable.subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer integer) throws Exception {
                       Utils.d("result： " + integer);
                }
            });



    }

    private void testEmpty() {

        Observable empty = Observable.empty();
//        Observable.empty()
        Observable errorable = Observable.error(new RuntimeException("程序出错咯 啊哈哈哈"));
        errorable.subscribe(new Consumer() {
              @Override
              public void accept(Object o) throws Exception {

              }
          }, new Consumer<Throwable>() {
              @Override
              public void accept(Throwable throwable) throws Exception {
                     Utils.d("throwable: " + throwable.getMessage());
              }
          }, new Action() {
              @Override
              public void run() throws Exception {
                     Utils.d("complete ");
              }
          });
    }


    private void tstConsumer() {


        Observable.just(1,2,3,4).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Utils.d(String.valueOf(integer));
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Utils.d("throwable" );

            }
        }, new Action() {
            @Override
            public void run() throws Exception {
                Utils.d("complete");
            }
        }, new Consumer<Disposable>() {
            @Override
            public void accept(Disposable disposable) throws Exception {
                  Utils.d("disposable:  " + disposable.isDisposed());
            }
        });
    }

    private void testJust() {
          Observable.just("hellow").subscribe(new Consumer<String>() {
              @Override
              public void accept(String s) throws Exception {
                     Utils.d("s");
              }
          });
    }

    private void testVoid() {

        Observable<String> observable =  Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                    e.onNext("One");
                    e.onNext("Twee");
                    e.onNext("Three");
                    e.onNext("Four");
                    e.onComplete();

            }
        });

        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
               Utils.d(s);
            }

            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
                 Utils.d("onCompelte");
            }
        };

//        observable.subscribe(observer);
        Subscriber<String> subscriber = new Subscriber<String>() {

            @Override
            public void onSubscribe(Subscription s) {

            }

            @Override
            public void onNext(String s) {
                Utils.d(s);
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {
                Utils.d("onCompelte");
            }
        };
//        observable.subscribe(subscriber);

//        Observable.s
    }

}
